package cn.kennylee.codehub.mq.lip.integration.kafka;

import cn.hutool.core.util.StrUtil;
import cn.kennylee.codehub.mq.lip.integration.kafka.listener.AbstractKafkaListener;
import cn.kennylee.codehub.mq.lip.integration.rocketmq.RocketMqListenerRegistrar;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.env.Environment;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * <p> 动态KafkaMqListener注册器 </p>
 * <p>Created on 2025/6/13.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Slf4j
@RequiredArgsConstructor
public class KafkaMqListenerRegistrar implements BeanPostProcessor, ApplicationContextAware {

    private ApplicationContext applicationContext;

    private final ConsumerFactory<String, String> consumerFactory;

    private Environment env;

    /**
     * BeanPostProcessor的后置处理方法，在Bean实例化后执行
     *
     * @param bean     the new bean instance
     * @param beanName the name of the bean
     * @return the bean instance to use, either the original or a wrapped one; never {@code null}
     */
    @Override
    public Object postProcessAfterInitialization(@NonNull Object bean, @NonNull String beanName) {
        // 只支持单例Bean，否则会重复注册
        if (!applicationContext.containsBean(beanName) ||
            !applicationContext.isSingleton(beanName)) {
            return bean;
        }

        if (bean instanceof AbstractKafkaListener abstractMqListener) {
            log.info("生成Kafka消费者容器: {}，消费者Channel: {}，消费者组: {}",
                abstractMqListener.getClass().getSimpleName(), abstractMqListener.getChannel(), addRouterKeySuffix(abstractMqListener.getConsumerGroup()));
            registerListenerContainer(abstractMqListener);
        }

        return bean;
    }

    /**
     * 注册Kafka消息监听容器
     *
     * @param listener 消息监听器
     */
    @NonNull
    public void registerListenerContainer(@NonNull AbstractKafkaListener listener) {
        BeanDefinitionBuilder builder = BeanDefinitionBuilder
            .rootBeanDefinition(ConcurrentMessageListenerContainer.class, () -> {
                Assert.notNull(listener.getChannel(), "Kafka Container Channel must not be null");

                // 1. 配置容器属性（指定Topic和监听器）
                ContainerProperties containerProps = new ContainerProperties(listener.getChannel());
                // TODO 设置分区或Header模拟Tag
                // 设置消费者组
                containerProps.setGroupId(addRouterKeySuffix(listener.getConsumerGroup()));
                // 设置消息转换器
                containerProps.setAckMode(ContainerProperties.AckMode.BATCH);
                // 绑定监听器
                containerProps.setMessageListener(listener);

                // 2. 创建容器并设置并发
                ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
                // 并发消费者数量
                container.setConcurrency(3);
                log.info("{}, 生成Kafka消费者容器注册成功: {}，消费者组: {}, 并发消费者数量: {}", listener.getClass().getSimpleName(),
                    containerProps.getTopics(), container.getGroupId(), container.getConcurrency());
                return container;
            });

        final String containerBeanName = StrUtil.format("{}_{}", ConcurrentMessageListenerContainer.class.getName(),
            listener.getInstanceName());
        log.info("注册Kafka Listener Container: {}", containerBeanName);

        ((DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory())
            .registerBeanDefinition(containerBeanName, builder.getBeanDefinition());
    }

    /**
     * 添加环境后缀
     *
     * @param s 目标字符串
     * @return 添加后缀的字符串
     */
    private String addRouterKeySuffix(String s) {
        return RocketMqListenerRegistrar.addRouterKeySuffix(s, applicationContext);
    }

    /**
     * 消费者组集合
     */
    private final Set<String> consumerGroups = ConcurrentHashMap.newKeySet();

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.env = applicationContext.getEnvironment();
    }
}
